package yz.mr.ReduceJoin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
 *  需求：
 *      读取学生数据的基本信息和学生的总分，在Reduce端与相同学生的数据进行拼接，形成join效果
 *
 *  Mapper
 *      ①读取传入的一行数据  1500100001,施笑槐,22,女,文科六班   1500100001	406
 *      ②对读取的数据进行判断，根据不同种类的数据分析处理
 *      ③以学生的id作为key  学生信息作为value
 *
 * Reduce
 *      ①接收相同学生的id，,施笑槐,22,女,文科六班 406
 *      ②对获取的信息进行判断，并且做拼接
 *      ③将结果数据写出   1500100001 做为key  其他的基本信息作为value
 */
public class ReduceJoin {
    public static void main(String[] args) throws Exception{

        //1.创建Job操作对象
        //2.创建配置类对象
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        //3.配置job
        job.setJobName("ReduceJoin");
        job.setJarByClass(ReduceJoin.class);

        //4.设置Mapper以及输出类型
        job.setMapperClass(ReduceJoinMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //5.设置Reduce以及输出对象
        job.setReducerClass(ReduceJoinReduce.class);

        //6.设置最终地输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //7.设置输入输出路径(这是本地的路径)  输入路径既可以指定路径也可以指定目录
        FileInputFormat.addInputPath(job,new Path("src/main/dataFile/students.txt"));
        FileInputFormat.addInputPath(job,new Path("output/scoreCount"));
        FileOutputFormat.setOutputPath(job,new Path("output/ReduceCount"));

        //8.提交执行当前JOB
        job.waitForCompletion(true);

    }
}
